package mapreduce;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.lang.reflect.Constructor;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.Vector;
import java.util.concurrent.CountDownLatch;
import org.apache.commons.collections.map.MultiValueMap;

public class Reducetask<KEYIN, VALUEIN, KEYOUT, VALUEOUT> extends Task {

	private ReduceTaskConf<KEYIN, VALUEIN, KEYOUT, VALUEOUT> reduceTaskConf;
	private Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> reducer;
	private CountDownLatch end;
	private CountDownLatch start;
    private boolean sortflag;
    private  Vector<pair<KEYIN, VALUEIN>>[][] map;
	public Reducetask(CountDownLatch start, CountDownLatch end, int taskid,
			ReduceTaskConf<KEYIN, VALUEIN, KEYOUT, VALUEOUT> conf, boolean flag,  Vector<pair<KEYIN, VALUEIN>>[][] middleBuffer) {
		super(taskid);
		this.reduceTaskConf = conf;
		this.end = end;
		this.start = start;
		this.sortflag=flag;
		this.map = middleBuffer;
	}

	@SuppressWarnings("unchecked")
	public void run() {
		try {
			end.await();
			MultiValueMap reduceMap = new MultiValueMap();
			String line;
			String[] lines;
			File dir = new File(reduceTaskConf.getFileDir());
			Class<?> reduceClass = reduceTaskConf.getReduceClass();
			Constructor<?> constructor = reduceClass.getConstructor();
			reducer = (Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>) constructor
					.newInstance();
			Object okey = null, ovalue = null;
			Class<?> inputKeyClass = reduceTaskConf.getInputKeyClass();
			Class<?> inputValueClass = reduceTaskConf.getInputValueClass();
			if(reduceTaskConf.getMemoryflag()==false)
			{for (File f : dir.listFiles()) {
				BufferedReader br = new BufferedReader(new FileReader(f));

				while ((line = br.readLine()) != null) {
					lines = line.split("\t");
					if (lines.length != 2)
						continue;
					if (inputKeyClass == Integer.class)
						okey = Integer.parseInt(lines[0]);
					else if (inputKeyClass == Double.class)
						okey = Double.parseDouble(lines[0]);
					else {
						okey = lines[0];
					}
					if (inputValueClass == Integer.class)
						ovalue = Integer.parseInt(lines[1]);
					else if (inputValueClass == Double.class)
						ovalue = Double.parseDouble(lines[1]);
					else {
						ovalue = lines[1];
					}
					reduceMap.put((KEYIN) okey, (VALUEIN) ovalue);
				}
			}
			}
			else {
				for( int i=0 ;i< map.length;i++)
				{
					for(pair<KEYIN, VALUEIN> p :map[i][TaskId])
					{
						reduceMap.put(p.getfirst(), p.getsecond());
					}
				}
			}
			if(this.sortflag==false)
			{
				Iterator<String> iterator = reduceMap.keySet().iterator();
				while (iterator.hasNext()) {

					okey = iterator.next();
					Collection<VALUEIN> values = (Collection<VALUEIN>) reduceMap
							.getCollection(okey);
					okey = (KEYIN) okey;
					reducer.reduce((KEYIN) okey, values, reduceTaskConf);

				}
			}
			else {
				KEYIN[] keys =(KEYIN[]) reduceMap.keySet().toArray();
				Arrays.sort(keys);
				for(KEYIN key: keys)
				{
					Collection<VALUEIN> values = (Collection<VALUEIN>) reduceMap
							.getCollection(key);
					reducer.reduce((KEYIN) key, values, reduceTaskConf);
				}
			}
			reduceTaskConf.close();
			start.countDown();
		} catch (Exception e) {
			// TODO: handle exception
		}

	}

}
